package com.jy.oms.framework.asyn.task.queue;

import com.alibaba.fastjson.JSON;
import com.jy.oms.framework.asyn.task.constants.TaskQueueTypeEnum;
import com.jy.oms.framework.asyn.task.constants.TaskStateEnum;
import com.jy.oms.framework.asyn.task.core.ExecutorConfig;
import com.jy.oms.framework.asyn.task.core.ExecutorContext;
import com.jy.oms.framework.asyn.task.exception.TaskConfigException;
import com.jy.oms.framework.asyn.task.executor.ITaskExecutor;
import com.jy.oms.framework.asyn.task.handler.ITaskHandlerManager;
import com.jy.oms.framework.asyn.task.handler.MethodTaskHandler;
import com.jy.oms.framework.asyn.task.log.ITaskLogService;
import com.jy.oms.framework.asyn.task.model.*;
import com.jy.oms.framework.asyn.task.queue.database.DbTaskQueueStatisticManager;
import com.jy.oms.framework.asyn.task.queue.memory.MemoryTaskQueueManager;
import com.jy.oms.framework.asyn.task.queue.redis.RedisTaskQueueManager;
import com.jy.oms.framework.asyn.task.util.AssertTask;
import com.jy.oms.framework.util.SpringUtil;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.tuple.Pair;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

public final class TaskQueueManagerProxy {
    private static Logger logger = LoggerFactory.getLogger(TaskQueueManagerProxy.class);

    public static ITaskQueueManager locate(String taskCode) {
        ITaskHandlerManager taskHandlerManager = SpringUtil.getBean(ITaskHandlerManager.class);
        TaskHandlerWrapper taskHandlerWrapper = taskHandlerManager.loadTaskHandler(taskCode);
        return locate(taskHandlerWrapper.getTaskQueue());
    }

    public static ITaskQueueManager locate(TaskQueueTypeEnum type) {
        if (TaskQueueTypeEnum.DB == type) {
            return SpringUtil.getBean(DbTaskQueueStatisticManager.class);
        } else if (TaskQueueTypeEnum.REDIS == type) {
            return SpringUtil.getBean(RedisTaskQueueManager.class);
        } else if (TaskQueueTypeEnum.MEM == type) {
            return SpringUtil.getBean(MemoryTaskQueueManager.class);
        }

        throw new TaskConfigException("TaskQueue[" + type + "] unsupported!");
    }

    /**
     * 添加任务到队列，任务将异步执行
     *
     * @param taskCode 任务编码
     * @param param    任务参数
     */
    public static void addTask(String taskCode, Object param) {
        TaskDto taskDto = new TaskDto();
        taskDto.setTaskCode(taskCode);
        taskDto.setParam(param);

        addTask(taskDto);
    }

    /**
     * 添加任务到队列，任务将异步执行
     *
     * @param taskDto 任务参数
     */
    public static void addTask(TaskDto taskDto) {
        valid(taskDto);

        ITaskHandlerManager taskHandlerManager = SpringUtil.getBean(ITaskHandlerManager.class);
        TaskHandlerWrapper taskHandlerWrapper = taskHandlerManager.loadTaskHandler(taskDto.getTaskCode());
        if (!taskHandlerWrapper.isEnableExecute()) {
            if (!taskHandlerWrapper.isEnableAddTaskIfUnableExecute()) {
                throw new TaskConfigException("任务不能运行，不能提交任务，任务编码：" + taskDto.getTaskCode()
                        + "，任务参数：" + JSON.toJSONString(taskDto.getParam()));
            }
        }

        TaskWrapper taskWrapper = convertTaskWrapper(taskDto, taskHandlerWrapper);
        try {
            locate(taskHandlerWrapper.getTaskQueue()).add(taskWrapper);
        } catch (Exception e) {
            logger.error("添加任务错误, task：" + JSON.toJSONString(taskDto), e);
        }
    }

    private static TaskWrapper convertTaskWrapper(TaskDto taskDto, TaskHandlerWrapper taskHandlerWrapper) {
        TaskWrapper taskWrapper = new TaskWrapper();
        taskWrapper.setTaskCode(taskDto.getTaskCode());
        taskWrapper.setTaskDto(taskDto);
        if (taskHandlerWrapper.isEnableExecute()) {
            taskWrapper.setTaskGroup(taskHandlerWrapper.getTaskGroup().getGroup());
        } else {
            taskWrapper.setTaskGroup(taskHandlerWrapper.getDefaultTaskGroupIfUnableExecute());
        }
        return taskWrapper;
    }

    /**
     * 批量添加任务到队列，任务将异步执行
     *
     * @param taskCode 任务编码
     * @param params   任务参数集合
     */
    public static void addTaskBatch(String taskCode, List<Object> params) {
        if (CollectionUtils.isEmpty(params)) {
            return;
        }

        List<TaskDto> taskDtoList = params.stream().map(param -> {
            TaskDto taskDto = new TaskDto();
            taskDto.setTaskCode(taskCode);
            taskDto.setParam(param);
            return taskDto;
        }).collect(Collectors.toList());

        addTaskBatch(taskDtoList);
    }

    /**
     * 批量添加任务到队列，任务将异步执行
     *
     * @param taskDtoList 任务参数
     */
    public static void addTaskBatch(List<TaskDto> taskDtoList) {
        if (CollectionUtils.isEmpty(taskDtoList)) {
            return;
        }

        Map<TaskQueueTypeEnum, List<TaskWrapper>> taskWrapperMap = new HashMap<>();
        ITaskHandlerManager taskHandlerManager = SpringUtil.getBean(ITaskHandlerManager.class);
        for (TaskDto taskDto : taskDtoList) {
            valid(taskDto);

            TaskHandlerWrapper taskHandlerWrapper = taskHandlerManager.loadTaskHandler(taskDto.getTaskCode());
            if (!taskHandlerWrapper.isEnableExecute()) {
                if (!taskHandlerWrapper.isEnableAddTaskIfUnableExecute()) {
                    throw new TaskConfigException("任务不能运行，不能提交任务，任务编码：" + taskDto.getTaskCode()
                            + "，任务参数：" + JSON.toJSONString(taskDto.getParam()));
                }
            }

            taskWrapperMap.computeIfAbsent(taskHandlerWrapper.getTaskQueue(), k -> new ArrayList<>());

            TaskWrapper taskWrapper = convertTaskWrapper(taskDto, taskHandlerWrapper);
            taskWrapperMap.get(taskHandlerWrapper.getTaskQueue()).add(taskWrapper);
        }

        for (Map.Entry<TaskQueueTypeEnum, List<TaskWrapper>> entry : taskWrapperMap.entrySet()) {
            try {
                locate(entry.getKey()).addBatch(entry.getValue());
            } catch (Exception e) {
                logger.error("添加任务错误, task：" + JSON.toJSONString(entry.getValue()), e);
            }
        }
    }

    /**
     * 任务存在则能添加任务，任务将异步执行
     *
     * @param taskCode 任务编码
     * @param param    任务参数
     */
    public static void addTaskIfExist(String taskCode, Object param) {
        TaskDto taskDto = new TaskDto();
        taskDto.setTaskCode(taskCode);
        taskDto.setParam(param);

        addTaskIfExist(taskDto);
    }

    /**
     * 任务存在则能添加任务，任务将异步执行
     *
     * @param taskDto 任务参数
     */
    public static void addTaskIfExist(TaskDto taskDto) {
        AssertTask.notNull(taskDto, "任务对象不能为空");
        AssertTask.notNull(taskDto.getTaskCode(), "任务编码不能为空");

        ITaskHandlerManager taskHandlerManager = SpringUtil.getBean(ITaskHandlerManager.class);
        TaskHandlerWrapper taskHandlerWrapper = taskHandlerManager.loadTaskHandler(taskDto.getTaskCode());
        if (taskHandlerWrapper == null) {
            return;
        }

        addTask(taskDto);
    }

    @SuppressWarnings("rawtypes")
    private static void valid(TaskDto taskDto) {
        AssertTask.notNull(taskDto, "任务对象不能为空");
        AssertTask.notNull(taskDto.getTaskCode(), "任务编码不能为空");
        AssertTask.notNull(taskDto.getParam(), "任务参数不能为空");

        ExecutorConfig executorConfig = ExecutorContext.getExecutorConfig();
        if (executorConfig == null || StringUtils.isAnyBlank(executorConfig.getBusinessGroup(), executorConfig.getEnv())) {
            throw new TaskConfigException("ExecutorConfig Bean没有定义BusinessGroup、Env属性");
        }

        ITaskHandlerManager taskHandlerManager = SpringUtil.getBean(ITaskHandlerManager.class);
        TaskHandlerWrapper taskHandlerWrapper = taskHandlerManager.loadTaskHandler(taskDto.getTaskCode());
        AssertTask.notNull(taskHandlerWrapper, "任务未注册，不能提交任务，任务编码：" + taskDto.getTaskCode());

        Class paramType = taskDto.getParam().getClass();
        Class taskHandlerParamType = taskHandlerWrapper.getTaskHandlerParamType();
        AssertTask.isTrue(paramType == taskHandlerParamType,
                "参数类型为[" + paramType.getName() + "]与任务参数类型[" + taskHandlerParamType.getName() + "]不匹配,"
                        + "任务编码：" + taskDto.getTaskCode());
    }

    /**
     * 立即执行任务，如果执行任务报错，则自动添加任务再异步重试
     *
     * @param taskCode 任务编码
     * @param param    任务参数
     */
    public static void execute(String taskCode, Object param) {
        TaskDto taskDto = new TaskDto();
        taskDto.setTaskCode(taskCode);
        taskDto.setParam(param);

        execute(taskDto);
    }

    /**
     * 立即执行任务，如果执行任务报错，则自动添加任务再异步重试
     *
     * @param taskDto 任务参数
     */
    public static boolean execute(TaskDto taskDto) {
        Pair<Boolean, Object> result = executeCommon(taskDto, true);
        return result.getLeft();
    }

    public static TaskExecuteResultDto executeAndGetResult(TaskDto taskDto) {
        Pair<Boolean, Object> result = executeCommon(taskDto, true);
        return new TaskExecuteResultDto(result.getLeft(), result.getRight());
    }

    /**
     * 立即执行任务，如果执行任务报错，只添加任务不重试
     *
     * @param taskDto 任务参数
     */
    public static boolean executeNeverRetryOnFailure(TaskDto taskDto) {
        Pair<Boolean, Object> result = executeCommon(taskDto, false);
        return result.getLeft();
    }

    public static TaskExecuteResultDto executeNeverRetryAndGetResultOnFailure(TaskDto taskDto) {
        Pair<Boolean, Object> result = executeCommon(taskDto, false);
        return new TaskExecuteResultDto(result.getLeft(), result.getRight());
    }

    private static Pair<Boolean, Object> executeCommon(TaskDto taskDto, boolean needRetry) {
        valid(taskDto);

        ITaskHandlerManager taskHandlerManager = SpringUtil.getBean(ITaskHandlerManager.class);
        TaskHandlerWrapper taskHandlerWrapper = taskHandlerManager.loadTaskHandler(taskDto.getTaskCode());
        TaskWrapper taskWrapper = convertTaskWrapper(taskDto, taskHandlerWrapper);
        taskWrapper.setCallCount(0);
        taskWrapper.setExecuteImmediate(true);
        taskWrapper.setExecuteImmediateRetryOnFailure(needRetry);

        ITaskExecutor taskExecutor = SpringUtil.getBean(ITaskExecutor.class);
        TaskResultDto taskResultDto = taskExecutor.executeTask(taskWrapper);
        if (taskResultDto.getTaskState() == TaskStateEnum.DONE) {
            return Pair.of(true, taskResultDto.getTaskResult());
        }

        //异常情况下补偿重试，MethodTaskHandler情况下，由切面自动插入重试任务数据
        if (taskHandlerWrapper.getTaskHandler().getClass() == MethodTaskHandler.class) {
            return Pair.of(false, null);
        }

        addTask(taskDto.setTaskState(needRetry ? TaskStateEnum.UNDO : TaskStateEnum.ERROR));

        return Pair.of(false, null);
    }

    /**
     * 从队列里删除任务
     *
     * @param taskCode 任务编码
     * @param param    任务参数
     */
    public static void removeTask(String taskCode, Object param) {
        TaskDto taskDto = new TaskDto();
        taskDto.setTaskCode(taskCode);
        taskDto.setParam(param);

        removeTask(taskDto);
    }

    /**
     * 从队列里删除任务
     *
     * @param taskDto 任务
     */
    public static void removeTask(TaskDto taskDto) {
        valid(taskDto);

        ITaskHandlerManager taskHandlerManager = SpringUtil.getBean(ITaskHandlerManager.class);
        TaskHandlerWrapper taskHandlerWrapper = taskHandlerManager.loadTaskHandler(taskDto.getTaskCode());
        TaskWrapper taskWrapper = convertTaskWrapper(taskDto, taskHandlerWrapper);
        try {
            locate(taskHandlerWrapper.getTaskQueue()).remove(taskWrapper);
        } catch (Exception e) {
            logger.error("删除任务错误, task：" + JSON.toJSONString(taskDto), e);
        }
    }

    /**
     * 添加任务日志
     *
     * @param taskLogDto 日志
     */
    public static void insertTaskLog(TaskLogDto taskLogDto) {
        ITaskLogService taskLogService = SpringUtil.getBean(ITaskLogService.class);
        taskLogService.insert(taskLogDto);
    }

    /**
     * 异步添加任务日志
     *
     * @param taskLogDto 日志
     */
    public static void insertTaskLogAsync(TaskLogDto taskLogDto) {
        ITaskLogService taskLogService = SpringUtil.getBean(ITaskLogService.class);
        taskLogService.insertAsync(taskLogDto);
    }
}
